package com.lichao.FinkDataProcess.sink;

import org.apache.flink.configuration.Configuration;

import java.io.Serializable;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class DynamicJdbcSink extends ParallelismFunction implements Serializable {

    private transient Connection connection;
    private transient PreparedStatement preparedStatement;
    private final String tableName;

    public DynamicJdbcSink(String tableName,int parelllelism) {
        super(parelllelism);
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/springboottest", "root", "123456");
        String sql = "INSERT INTO " + tableName + " (url) VALUES (?)";
        preparedStatement = connection.prepareStatement(sql);
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        preparedStatement.setString(1, value);
        preparedStatement.executeUpdate();
    }

    @Override
    public void close() throws Exception {
        if (preparedStatement != null) {
            preparedStatement.close();
        }
        if (connection != null) {
            connection.close();
        }
        super.close();
    }
}